package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}
type KeyValueSlice []KeyValue

func (keyValueSlice KeyValueSlice) Len() int {
	return len(keyValueSlice)
}
func (keyValueSlice KeyValueSlice) Less(i, j int) bool {
	return keyValueSlice[i].Key < keyValueSlice[j].Key
}
func (keyValueSlice KeyValueSlice) Swap(i, j int) {
	keyValueSlice[i], keyValueSlice[j] = keyValueSlice[j], keyValueSlice[i]
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	// Your worker implementation here.
	keepFlag := true
	for keepFlag {
		task := getTask()
		switch task.TaskType {
		case MapTask:
			{
				doMapTask(task, mapf)
			}
		case ReduceTask:
			{
				doReduceTask(task, reducef)
			}
		case WaitingTask:
			{
				time.Sleep(time.Second)
			}
		case ExitTask:
			{
				keepFlag = false
			}
		}
	}

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()

}

//map处理
func doMapTask(task Task, mapf func(string, string) []KeyValue) {
	filename := task.FileName
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filename)
	}
	file.Close()
	//map处理完后返回key-value键值对数组
	kva := mapf(filename, string(content))
	rn := task.ReducerNum
	//创建一个key-value二维数组
	HashedKV := make([][]KeyValue, rn)
	for _, kv := range kva {
		HashedKV[ihash(kv.Key)%rn] = append(HashedKV[ihash(kv.Key)%rn], kv)
	}
	//将reduce需要处理的内容写入到临时文件
	for i := 0; i < rn; i++ {
		tempName := "mr-tmp-" + strconv.Itoa(task.TaskId) + "-" + strconv.Itoa(i)
		file, err := os.Create(tempName)
		if err != nil {
			println("create file err")
			return
		}
		enc := json.NewEncoder(file)
		for _, kv := range HashedKV[i] {
			enc.Encode(kv)
		}
		file.Close()
	}
	taskDone(task)
}

//调用rpc向协调者发送请求获取任务
func getTask() Task {
	args := TaskArgs{}
	reply := Task{}
	ok := call("Coordinator.PollTask", &args, &reply)
	if ok {
		fmt.Printf("reply: %v\n", reply)
	} else {
		log.Fatal("call error")
	}
	return reply
}

//调用rpc向协调者发送请求更新任务状态
func taskDone(task Task) {
	reply := Task{}
	ok := call("Coordinator.TaskDone", task, &reply)
	if ok {
		fmt.Printf("call task[%d] done ok \n", task.TaskId)
	} else {
		fmt.Println("call failed!")
	}
}

//reduce处理
func doReduceTask(task Task, reducef func(string, []string) string) {
	filesName := task.FileSlice
	// fmt.Println(task.TaskId, ":", task.FileSlice)
	//读取该任务要处理的所有key-value
	keyvalues := shuffleFile(filesName)
	i := 0
	// 创建一个临时文件
	tmpFile, err := ioutil.TempFile(os.TempDir(), "mr-tmp-*")
	if err != nil {
		fmt.Println(err)
	}
	for i < len(keyvalues) {
		j := i + 1
		for j < len(keyvalues) && keyvalues[i].Key == keyvalues[j].Key {
			j++
		}
		//从i到j的key-value的key都是一样的
		keyvalue := make([]string, 0)
		for k := i; k < j; k++ {
			keyvalue = append(keyvalue, keyvalues[k].Value)
		}
		v := reducef(keyvalues[i].Key, keyvalue)
		fmt.Fprintf(tmpFile, "%v %v\n", keyvalues[i].Key, v)
		i = j
	}

	tmpFile.Close()
	fn := fmt.Sprintf("mr-out-%d", task.TaskId)
	os.Rename(tmpFile.Name(), fn)
	taskDone(task)
}

//reduce阶段读取文件得到每个reduce需要处理的key-value
func shuffleFile(filesName []string) []KeyValue {
	kva := make([]KeyValue, 0)
	for _, fileName := range filesName {
		file, err := os.Open(fileName)
		if err != nil {
			fmt.Println(err)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
	}
	sort.Sort(KeyValueSlice(kva))
	return kva
}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
